package bch.rocketmq.service;// OrderDelayedQueueService.java

import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Service
@Slf4j
public class OrderDelayedQueueService implements SmartLifecycle {

    private static final String QUEUE_NAME = "order_delayed_queue";

    private final AtomicBoolean running = new AtomicBoolean(false);

    @Autowired
    @Qualifier("delayedQueueExecutor")
    private ThreadPoolTaskExecutor delayedQueueExecutor;
    @Autowired
    private RedissonClient redissonClient;

    private RBlockingDeque<String> blockingDeque;
    private RDelayedQueue<String> delayedQueue;
    private RBlockingDeque<String> processingQueue;


    /**
     * 取出任务并标记为处理中（原子操作）
     */
    private String fetchAndMarkPending() {
        String lua = """
                local val = redis.call('LPOP', KEYS[1]) 
                if val then 
                   redis.call('RPUSH', KEYS[2], val) 
                end 
                return val
                """;

        return redissonClient.getScript().eval(
                org.redisson.api.RScript.Mode.READ_WRITE,
                lua,
                org.redisson.api.RScript.ReturnType.VALUE,
                java.util.Arrays.asList(QUEUE_NAME, "order_delayed_queue_processing")
        );
    }

    /**
     * 标记任务已完成，移出处理中队列
     */
    private void markTaskDone(String orderId) {
        if (orderId != null) {
            boolean remove = processingQueue.remove(orderId);
            if (remove) {
                log.info("订单 {} 已完成。", orderId);
            }
        }
    }

    @PostConstruct
    public void init() {
        // 获取一个阻塞队列，作为延迟队列的目标队列
        this.blockingDeque = redissonClient.getBlockingDeque(QUEUE_NAME);
        // 创建一个延迟队列，并将其与阻塞队列关联
        this.delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        this.processingQueue = redissonClient.getBlockingDeque(QUEUE_NAME + "_processing");
        log.info("延迟队列已初始化。");
//        startConsumerTasks();
        // 启动消费者线程, 使用SmartLifecycle接口
//        startConsumer();
    }

    /**
     * 添加任务到延迟队列
     *
     * @param orderId 订单ID
     * @param delay   延迟时间
     * @param unit    时间单位
     */
    public void addOrder(String orderId, long delay, TimeUnit unit) {
        delayedQueue.offer(orderId, delay, unit);
    }

    /**
     * 启动消费者线程，持续从阻塞队列中获取到期任务
     */
    private void startConsumer() {
        delayedQueueExecutor.execute(() -> {
            log.info("延迟队列消费者已启动，等待任务...");
            while (running.get() && !Thread.currentThread().isInterrupted()) {
                try {
                    // 没有竟态条件
                    // 当一个新元素进入队列时，Redis 内部的机制会确保它只会被一个阻塞的客户端获取
                    // 异常 记录并落库 任务重新放回队列, 如果取出来了执行过程中挂了 任务不就没了吗

                    // 优雅停止
                    String orderId = fetchAndMarkPending(); // 原子取出 + 标记处理中
                    // 放入预留队列 (RBlockingDeque)：存储正在被处理中的任务，并记录处理开始时间
                    // 如果极端情况take出来还未放入预留队列，服务挂掉了 那么使用lua脚本即可
                    if (orderId != null) {
                        log.info("开始处理订单: {}", orderId);
                        // 模拟业务处理
                        Thread.sleep(100);

                        // 处理完成，移出 processing 队列
                        markTaskDone(orderId);
                    }
                } catch (InterruptedException e) {
                    // 恢复中断标志
                    Thread.currentThread().interrupt();
                    log.error("消费者线程被中断。");
                    break;
                }
            }
            log.info("延迟队列消费者已安全退出。");
        });
    }


    private void startConsumerTasks() {
        // 可以启动多个线程消费
        int consumerCount = 2; // 示例：2个线程
        for (int i = 0; i < consumerCount; i++) {
            delayedQueueExecutor.execute(() -> {
                while (running.get() && !Thread.currentThread().isInterrupted()) {
                    try {
                        String orderId = fetchAndMarkPending();
                        if (orderId != null) {
                            log.info("开始处理订单: {}", orderId);
                            Thread.sleep(100); // 模拟业务处理
                            markTaskDone(orderId);
                        } else {
                            // 没有任务，短暂休眠
                            Thread.sleep(200);
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        log.warn("消费者线程被中断");
                        break;
                    } catch (Exception e) {
                        log.error("处理订单异常", e);
                    }
                }
                log.info("消费者线程安全退出");
            });
        }
    }

    /** SmartLifecycle 接口实现 **/

    @Override
    public void start() {
        if (running.compareAndSet(false, true)) {
            startConsumerTasks();
            log.info("启动延迟队列消费者容器");
        }
    }



    @Override
    public void stop() {
        stop(() -> {});
    }

    @Override
    public void stop(Runnable callback) {
        if (running.compareAndSet(true, false)) {
            log.info("停止延迟队列消费者容器，等待任务完成...");
            ExecutorService executor = delayedQueueExecutor.getThreadPoolExecutor();
            try {
                // 优雅停机模板
                executor.shutdown(); // 不接收新任务
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    executor.shutdownNow(); // 超时中断阻塞任务
                    if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                        log.warn("延迟队列消费者线程池未能完全退出");
                    }
                }
            } catch (InterruptedException e) {
                executor.shutdownNow();
                Thread.currentThread().interrupt();
            }
            log.info("延迟队列消费者容器已停止");
            callback.run();
        }
    }

    @Override
    public boolean isRunning() {
        return running.get();
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public int getPhase() {
        return Integer.MAX_VALUE; // 延迟停止，保证先停止 web 请求
    }
}